import org.apache.commons.collections.CollectionUtils;
import org.apache.commons.lang3.StringUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.CellUtil;
import org.apache.hadoop.hbase.CompareOperator;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.HColumnDescriptor;
import org.apache.hadoop.hbase.HTableDescriptor;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.Admin;
import org.apache.hadoop.hbase.client.Connection;
import org.apache.hadoop.hbase.client.ConnectionFactory;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.ResultScanner;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.client.Table;
import org.apache.hadoop.hbase.filter.FilterList;
import org.apache.hadoop.hbase.filter.SingleColumnValueFilter;
import org.apache.hadoop.hbase.filter.SubstringComparator;
import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
import org.apache.hadoop.hbase.mapred.TableOutputFormat;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.mapred.JobConf;
import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import scala.Tuple2;

import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Scanner;
import java.util.TreeMap;
import java.util.regex.Matcher;
import java.util.regex.Pattern;

public class LogAnalyze {
    public static final byte[] FAMILY_NAME = Bytes.toBytes("infos");      // 列族

    public final static byte[] VISIT_TIME = Bytes.toBytes("visit_time");  // 访问时间
    public final static byte[] U_ID = Bytes.toBytes("u_id");              // 用户ID
    public final static byte[] QUERY_WORD = Bytes.toBytes("query_word");  // [查询词]
    public final static byte[] RANK = Bytes.toBytes("rank");              // 该URL在返回结果中的排名
    public final static byte[] ORDINAL = Bytes.toBytes("ordinal");        // 用户点击的顺序号
    public final static byte[] URL = Bytes.toBytes("url");                // 用户点击的URL

    // 数组 ["访问时间", "用户ID", "[查询词]", "该URL在返回结果中的排名", "用户点击的顺序号", "用户点击的URL"] 的下标
    public final static int VISIT_TIME_INDEX = 0;   // 访问时间
    public final static int U_ID_INDEX = 1;         // 用户ID
    public final static int QUERY_WORD_INDEX = 2;   // [查询词]
    public final static int RANK_INDEX = 3;         // 该URL在返回结果中的排名
    public final static int ORDINAL_INDEX = 4;      // 用户点击的顺序号
    public final static int URL_INDEX = 5;          // 用户点击的URL

    // 配置信息
    private static String path = "file:///home/hadoop/log_analyze/SogouQ.sample";
    private final static String tableNameStr = "LogAnalyze";
    private static final String hbaseRootDir = "hdfs://localhost:9000/hbase";

    private static final JavaSparkContext sc;

    static {
        SparkConf sparkConf = new SparkConf()
                .setMaster("local")
                .setAppName("SparkApp");
        sc = new JavaSparkContext(sparkConf);
    }

    public static void main(String[] args) throws IOException {
        int opt;
        Scanner scanner = new Scanner(System.in);
        LogAnalyze logAnalyze = new LogAnalyze();

        while (true) {
            System.out.print(
                    "===========================================================================\n" +
                            "* 1: 从默认文件中加载数据至HBase;\n" +
                            "* 11: 从指定文件中加载数据至HBase;\n" +
                            "* 2: 支持根据六个字段的一个或多个字段内容，使用不同条件（关键字或数值范围）进行联合搜索\n" +
                            "* 3: RDD联合搜索(功能和2一样)\n" +
                            "* 4: 时段流量统计: 以hh:mm:ss格式输入起始时间和结束时间，统计这段时间之内的总搜索次数、各个查询词搜索次数，各个网站的访问量\n" +
                            "* 5: 用户使用频率统计: 统计每个用户一天内的搜索次数\n" +
                            "* 6: 访问行为统计: 根据该页面在搜索结果中的排名，统计不同排名的结果被访问的情况\n" +
                            "* 0: 退出\n" +
                            "===========================================================================\n" +
                            ">"
            );
            opt = scanner.nextInt();
            if (opt == 0) {
                break;
            }
            switch (opt) {
                case 1: {
                    createTable(tableNameStr, new String[]{new String(LogAnalyze.FAMILY_NAME)}, true);
                    logAnalyze.file2Hbase();
                    System.out.println("加载完成！！！");
                    break;
                }
                case 11: {
                    System.out.print("请输入文件名\n>");
                    String input = scanner.next();
                    if (input.length() > 5) {
                        path = input.substring(5);
                    }
                    createTable(tableNameStr, new String[]{new String(LogAnalyze.FAMILY_NAME)}, true);
                    logAnalyze.file2Hbase();
                    System.out.println("加载完成！！！");
                    break;
                }
                case 2: {
                    System.out.print("请输入查询条件([startTime][|endTime]+[userID]+[keyword]+[URL])\n>");
                    String input = scanner.next();
                    List<String> list = logAnalyze.scan(AppUtil.getQueryTerms(input));
                    list.forEach(System.out::println);
                    System.out.println("total: " + list.size());
                    break;
                }
                case 3: {
                    System.out.print("请输入查询条件([startTime][|endTime]+[userID]+[keyword]+[URL])\n>");
                    String input = scanner.next();
                    List<String> list = logAnalyze.queryByRDD(AppUtil.getQueryTerms(input));
                    list.forEach(System.out::println);
                    System.out.println("total: " + list.size());
                    break;
                }
                case 4: {
                    System.out.print("请输入以hh:mm:ss格式输入起始时间和结束时间(startTime|endTime)\n>");
                    String input = scanner.next();
                    QueryTerms queryTerms = AppUtil.getStartTimeEndTime(input);
                    System.out.println(queryTerms);
                    Map<String, Object> result = logAnalyze.flowStatisticsByTime(queryTerms.getStartTime(), queryTerms.getEndTime());
                    System.out.println(AppUtil.obj2Json(result));
                    break;
                }
                case 5: {
                    Map<String, Long> countByUser = logAnalyze.countByUser();
                    System.out.println(AppUtil.obj2Json(countByUser));
                    break;
                }
                case 6: {
                    Map<String, Long> statisticsMap = logAnalyze.accessBehaviorStatistics();
                    System.out.println(AppUtil.obj2Json(statisticsMap));
                    break;
                }
            }
        }
    }

    /**
     * 创建表，要求当 HBase 已经存在名为 tableName 的表的时候，先删除原有的表，再创建新的表。
     *
     * @param tableName 表的名称
     * @param fields    存储记录各个域名称的数组
     * @param isCover   是否覆盖
     */
    public static void createTable(String tableName, String[] fields, boolean isCover) {
        TableName createTableName = TableName.valueOf(tableName);
        Configuration hbConf = HBaseConfiguration.create();
        hbConf.set("hbase.rootdir", hbaseRootDir);
        try (Connection hbConn = ConnectionFactory.createConnection(hbConf)) {
            Admin admin = hbConn.getAdmin();
            if (admin.tableExists(createTableName)) {
                System.out.println("table is exists!");
                if (isCover) {
                    admin.disableTable(createTableName);
                    admin.deleteTable(createTableName);
                } else {
                    return;
                }
            }
            HTableDescriptor hTableDescriptor = new HTableDescriptor(createTableName);
            for (String field : fields) {
                hTableDescriptor.addFamily(new HColumnDescriptor(field));
            }
            admin.createTable(hTableDescriptor);
            System.out.println("----------------\ncreate table ok!");
            admin.close();
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    /**
     * （1）数据清洗部分：实现一个程序，载入文本中的日志数据，
     * 每行作为一条记录，分为访问时间，用户ID，查询词，返回结果排名，顺序号，URL这六个字段（列），
     * 存入HBASE。
     */
    public void file2Hbase() {
        // 1、spark 读取数据、清洗数据
        // 数据格式：访问时间 用户ID    [查询词]   该URL在返回结果中的排名   用户点击的顺序号    用户点击的URL
        JavaRDD<String[]> rdd = sc.textFile(path).map(s -> s.split("\\s"));

        // 2、HBase Configuration
        Configuration hbConf = HBaseConfiguration.create();
        hbConf.set("hbase.rootdir", hbaseRootDir);

        // 3、JobConf
        JobConf jobConf = new JobConf(hbConf, this.getClass());
        jobConf.setOutputFormat(TableOutputFormat.class);
        jobConf.set(TableOutputFormat.OUTPUT_TABLE, tableNameStr);

        // 4、rdd -> pairRDD
        JavaPairRDD<ImmutableBytesWritable, Put> pairRDD = rdd.mapToPair(line -> {
            byte[] rowKey = Bytes.toBytes(AppUtil.getUUID());
            ImmutableBytesWritable writable = new ImmutableBytesWritable(rowKey);
            Put put = new Put(rowKey);
            put.addColumn(FAMILY_NAME, VISIT_TIME, Bytes.toBytes(line[VISIT_TIME_INDEX]));
            put.addColumn(FAMILY_NAME, U_ID, Bytes.toBytes(line[U_ID_INDEX]));
            put.addColumn(FAMILY_NAME, QUERY_WORD, Bytes.toBytes(line[QUERY_WORD_INDEX]));
            put.addColumn(FAMILY_NAME, RANK, Bytes.toBytes(line[RANK_INDEX]));
            put.addColumn(FAMILY_NAME, ORDINAL, Bytes.toBytes(line[ORDINAL_INDEX]));
            put.addColumn(FAMILY_NAME, URL, Bytes.toBytes(line[URL_INDEX]));

            return new Tuple2<>(writable, put);
        });

        // 5、save as Hbase
        pairRDD.saveAsHadoopDataset(jobConf);
    }

    /**
     * （2）数据搜索部分:
     * 使用HBASE的API实现条件查询功能，例如: “访问时间在1点到2点之间且URL属于百度的网页”等，
     * 即能够支持根据六个字段的一个或多个字段内容，使用不同条件（关键字或数值范围）进行联合搜索。具体内容如下:
     * （1）根据`开始时间`和`结束时间`搜索这段时间内的访问记录。输入信息为开始时间和结束时间，用‘`|`’字符隔开）
     * （2）根据`用户ID`搜索该用户的访问记录。输入信息为一个或多个用户ID，用‘`|`’字符隔开
     * （输入多个用户ID时，要求记录匹配其中任意一个用户的均满足条件）。
     * （3）根据关键字搜索含有该关键字的`用户查询词记录`。输入信息为一个或多个关键字，用‘`|`’字符隔开
     * （输入多个关键字时，要求记录匹配其中任意一个关键字的均满足条件）。
     * （4）根据关键字（如baidu）等搜索相关网站的`URL`访问记录，输入信息为一个或多个关键字，用‘`|`’字符隔开
     * （输入多个关键字时，要求记录匹配其中任意一个关键字的均满足条件）。
     * （5）实现以上四个条件任意组合的联合搜索。四个条件之间用’+’字符隔开 （联合搜索要求所有条件同时满足）
     *
     * @param queryTerms 查询条件
     * @return 查询结果集
     * @throws IOException getTable异常
     */
    public List<String> scan(QueryTerms queryTerms) throws IOException {
        // hbase table
        Configuration hbConf = HBaseConfiguration.create();
        hbConf.set("hbase.rootdir", hbaseRootDir);
        Connection hbConn = ConnectionFactory.createConnection(hbConf);
        Table table = hbConn.getTable(TableName.valueOf(tableNameStr));

        Scan scan = new Scan();

        FilterList filterList = new FilterList(FilterList.Operator.MUST_PASS_ALL); // AND

        if (StringUtils.isNotBlank(queryTerms.getStartTime())) {// infos.visit_time >= startTime
            filterList.addFilter(new SingleColumnValueFilter(FAMILY_NAME, VISIT_TIME,
                    CompareOperator.GREATER_OR_EQUAL, Bytes.toBytes(queryTerms.getStartTime())));
        }
        if (StringUtils.isNotBlank(queryTerms.getEndTime())) {// infos.visit_time <= endTime
            filterList.addFilter(new SingleColumnValueFilter(FAMILY_NAME, VISIT_TIME,
                    CompareOperator.LESS_OR_EQUAL, Bytes.toBytes(queryTerms.getEndTime())));
        }

        if (CollectionUtils.isNotEmpty(queryTerms.getUserIds())) {// infos.u_id in userIds
            FilterList uidFilterList = new FilterList(FilterList.Operator.MUST_PASS_ONE); // OR
            for (String uid : queryTerms.getUserIds()) {
                uidFilterList.addFilter(new SingleColumnValueFilter(FAMILY_NAME, U_ID,
                        CompareOperator.EQUAL, new SubstringComparator(uid)));
            }
            filterList.addFilter(uidFilterList);
        }

        if (CollectionUtils.isNotEmpty(queryTerms.getKeyWords())) {// infos.query_word in keywords
            FilterList kwFilterList = new FilterList(FilterList.Operator.MUST_PASS_ONE);
            for (String kw : queryTerms.getKeyWords()) {
                kwFilterList.addFilter(new SingleColumnValueFilter(FAMILY_NAME, QUERY_WORD,
                        CompareOperator.EQUAL, new SubstringComparator(kw)));
            }
            filterList.addFilter(kwFilterList);
        }

        if (CollectionUtils.isNotEmpty(queryTerms.getUrls())) {// infos.url in urls
            FilterList kwFilterList = new FilterList(FilterList.Operator.MUST_PASS_ONE);
            for (String url : queryTerms.getUrls()) {
                kwFilterList.addFilter(new SingleColumnValueFilter(FAMILY_NAME, URL,
                        CompareOperator.EQUAL, new SubstringComparator(url)));
            }
            filterList.addFilter(kwFilterList);
        }
        scan.setFilter(filterList);
        System.out.println(scan.toJSON());
        List<String> resultList = new ArrayList<>(16);
        try (ResultScanner results = table.getScanner(scan)) {
            for (Result result : results) {
                resultList.add(String.format("[%s, %s, %s, %s, %s, %s]",
                        new String(CellUtil.cloneValue(result.getColumnCells(FAMILY_NAME, VISIT_TIME).get(0)), StandardCharsets.UTF_8),
                        new String(CellUtil.cloneValue(result.getColumnCells(FAMILY_NAME, U_ID).get(0)), StandardCharsets.UTF_8),
                        new String(CellUtil.cloneValue(result.getColumnCells(FAMILY_NAME, QUERY_WORD).get(0)), StandardCharsets.UTF_8),
                        new String(CellUtil.cloneValue(result.getColumnCells(FAMILY_NAME, RANK).get(0)), StandardCharsets.UTF_8),
                        new String(CellUtil.cloneValue(result.getColumnCells(FAMILY_NAME, ORDINAL).get(0)), StandardCharsets.UTF_8),
                        new String(CellUtil.cloneValue(result.getColumnCells(FAMILY_NAME, URL).get(0)), StandardCharsets.UTF_8)
                ));
            }
        }
        return resultList;
    }

    /**
     * 基于大数据计算技术的条件查询: 使用RDD算子，实现类似于HBASE的六个字段条件搜索。
     * 功能同上 SogoLog#scan(query.Query)
     *
     * @param queryTerms 查询条件
     * @return 查询结果集
     */
    public List<String> queryByRDD(QueryTerms queryTerms) {
        JavaRDD<String[]> rdd = sc.textFile(path).map(s -> s.split("\\s"));

        if (StringUtils.isNotBlank(queryTerms.getStartTime())) {// infos.visit_time >= startTime
            rdd = rdd.filter(line -> line[VISIT_TIME_INDEX].compareTo(queryTerms.getStartTime()) >= 0);
        }
        if (StringUtils.isNotBlank(queryTerms.getEndTime())) {// infos.visit_time <= endTime
            rdd = rdd.filter(line -> line[VISIT_TIME_INDEX].compareTo(queryTerms.getEndTime()) <= 0);
        }

        if (CollectionUtils.isNotEmpty(queryTerms.getUserIds())) {// infos.u_id in userIds
            rdd = rdd.filter(line -> {
                for (String uid : queryTerms.getUserIds()) {
                    if (line[U_ID_INDEX].contains(uid)) {
                        return true;
                    }
                }
                return false;
            });
        }

        if (CollectionUtils.isNotEmpty(queryTerms.getKeyWords())) {// infos.query_word in keywords
            rdd = rdd.filter(line -> {
                for (String kw : queryTerms.getKeyWords()) {
                    if (line[QUERY_WORD_INDEX].contains(kw)) {
                        return true;
                    }
                }
                return false;
            });
        }

        if (CollectionUtils.isNotEmpty(queryTerms.getUrls())) {// infos.url in urls
            rdd = rdd.filter(line -> {
                for (String url : queryTerms.getUrls()) {
                    if (line[URL_INDEX].contains(url)) {
                        return true;
                    }
                }
                return false;
            });
        }

        final long[] id = {1L};
        JavaPairRDD<Long, String> pairRDD = rdd.mapToPair(lines -> {
            id[0] = id[0] + 1L;
            return new Tuple2<>(id[0], Arrays.toString(lines));
        });
        Map<Long, String> collectAsMap = pairRDD.collectAsMap();

        return new ArrayList<>(collectAsMap.values());
    }

    /**
     * 时段流量统计: 以`hh:mm:ss`格式输入`起始时间`和`结束时间`，统计这段时间之内的`总搜索次数`、`各个查询词搜索次数`，`各个网站的访问量`。
     * 其中网站访问量为根据URL统计网站访问，属于同一个网站的URL算在同一个网站上，
     * 如 women.sohu.com/20070508/n249762812.shtml，
     * s.sohu.com/20080220/n255256097.shtml，
     * peng.blog.sohu.com等
     * 不同的sohu网页都统计为sohu网站的访问次数。
     *
     * @param startTime 起始时间
     * @param endTime   结束时间
     * @return 段时间之内的`总搜索次数`、`各个查询词搜索次数`，`各个网站的访问量`
     */
    public Map<String, Object> flowStatisticsByTime(String startTime, String endTime) {
        Map<String, Object> result = new HashMap<>(3);
        JavaRDD<String> stringJavaRDD = sc.textFile(path);
        JavaRDD<String[]> rdd = stringJavaRDD.map(s -> s.split("\\s"));
        rdd = rdd.filter(line -> line[VISIT_TIME_INDEX].compareTo(startTime) >= 0 && line[VISIT_TIME_INDEX].compareTo(endTime) <= 0);

        long allCount = rdd.count(); // 总搜索次数
        result.put("search", allCount);

        Map<String, Long> searchCount = rdd
                .keyBy(line -> line[QUERY_WORD_INDEX])
                .countByKey(); // 各个查询词搜索次数
        result.put("searchCount", searchCount);

        Pattern p = Pattern.compile("[^.]*?\\.(com.cn|net.cn|org.cn|gov.cn|edu.cn|com|cn|cc|tv|biz|edu|gov|info|int|mil|name|net|org|pro|xyz)", Pattern.CASE_INSENSITIVE);

        rdd = rdd.map(line -> {
            String[] split = line[URL_INDEX].split("/");
            if (split.length > 0) {
                line[URL_INDEX] = split[0];
            }
            Matcher m = p.matcher(line[URL_INDEX]);
            if (m.find()) {
                line[URL_INDEX] = m.group();
            }
            return line;
        }); // 提取一级域名

        Map<String, Long> visitCount = rdd
                .keyBy(line -> line[URL_INDEX])
                .countByKey(); // 各个网站的访问量

        result.put("visitCount", visitCount);

        return result;
    }

    /**
     * 用户使用频率统计: 统计每个用户一天内的搜索次数
     *
     * @return 统计每个用户一天内的搜索次数
     */
    public Map<String, Long> countByUser() {
        JavaRDD<String[]> rdd = sc.textFile(path)
                .map(s -> s.split("\\s"));

        return new TreeMap<>(rdd.keyBy(line -> line[U_ID_INDEX]).countByKey());
    }

    /**
     * 访问行为统计: 根据该页面在搜索结果中的排名(第4字段)，统计不同排名的结果被访问的情况。
     * 如排名第一的结果被访问次数，排名第二的结果被访问次数……
     *
     * @return 统计不同排名的结果被访问的情况
     */
    public Map<String, Long> accessBehaviorStatistics() {
        JavaRDD<String[]> rdd = sc.textFile(path)
                .map(s -> s.split("\\s"));

        return new TreeMap<>(rdd
                .keyBy(line -> line[QUERY_WORD_INDEX] + "=>" + line[RANK_INDEX])
                .countByKey());
    }

}
